import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.cep.pattern.Pattern;

import java.util.List;
import java.util.Map;


/**
 * @author wangzj
 * @description 复杂事件处理
 * 模拟电商网站用户搜索的数据来作为数据的输入源，
 * 然后查找其中重复搜索某一个商品的人，并且发送一条告警消息。
 * @date 2020/7/30 0:21
 */
public class FlinkCEPDemo {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //1、定义数据源
        DataStreamSource source = env.fromElements(
                //浏览记录
                Tuple3.of("Marry", "外套", 1L),
                Tuple3.of("Marry", "帽子", 1L),
                Tuple3.of("Marry", "帽子", 2L),
                Tuple3.of("Marry", "帽子", 3L),
                Tuple3.of("Ming", "衣服", 1L),
                Tuple3.of("Marry", "鞋子", 1L),
                Tuple3.of("Marry", "鞋子", 2L),
                Tuple3.of("LiLei", "帽子", 1L),
                Tuple3.of("LiLei", "鞋子", 2L),
                Tuple3.of("LiLei", "帽子", 3L),
                Tuple3.of("LiLei", "鞋子", 4L),
                Tuple3.of("LiLei", "帽子", 5L),
                Tuple3.of("LiLei", "帽子", 6L)
        );
        //2、定义Pattern,寻找连续搜索帽子的用户
        Pattern<Tuple3<String, String, Long>, Tuple3<String, String, Long>> pattern = Pattern
                .<Tuple3<String, String, Long>>begin("start")
                .where(new SimpleCondition<Tuple3<String, String, Long>>() {
                    @Override
                    public boolean filter(Tuple3<String, String, Long> value) throws Exception {
                        //返回访问帽子的用户
                        return value.f1.equals("帽子");
                    }
                })
                //前面的模式重复了3次
                .timesOrMore(3)
                //middle标识第二次访问了帽子
                .next("middle")
                .where(new SimpleCondition<Tuple3<String, String, Long>>() {
                    @Override
                    public boolean filter(Tuple3<String, String, Long> value) throws Exception {
                        return value.f1.equals("帽子");
                    }
                });

        //获取第一个元素作为key
        KeyedStream keyedStream = source.keyBy(0);
        //将pattern转成patternStream
        PatternStream patternStream = CEP.pattern(keyedStream, pattern);
        //3、将PatternStream转换成DataSream
        SingleOutputStreamOperator matchStream = patternStream.select(new PatternSelectFunction<Tuple3<String, String, Long>, String>() {
            @Override
            public String select(Map<String, List<Tuple3<String, String, Long>>> pattern) throws Exception {
                List<Tuple3<String, String, Long>> middle = pattern.get("middle");
                return middle.get(0).f0 + ":" + middle.get(0).f2 + ":" + "连续搜索了两次帽子";
            }
        });
        matchStream.printToErr();
        env.execute("execute cep");
    }
}
